package com.xzc.apitest.source

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka._

import scala.util.Random

//定义样例类，温度传感器
case class SensorReading(id: String, timestamp: Long, temperature: Double)

object SourceTest {
  def main(args: Array[String]): Unit = {
    //创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //1-从集合中读取数据
    val dataList = List(
      SensorReading("_1", 1547718199, 35.8),
      SensorReading("_6", 1547718201, 15.8),
      SensorReading("_7", 1547718202, 6.8),
      SensorReading("_10", 1547718300, 38.8)
    )
    val stream1 = env.fromCollection(dataList)
    //    stream1.print()

    //2-文件
    val stream2 = env.readTextFile("D:\\git\\learning_flink\\_01_试用\\src\\main\\resources\\sensor.txt")
    //    stream2.print()

    //3-kafka
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "hadoop102:9092")
    properties.setProperty("group.id", "flink-group")
    val stream3 = env.addSource(new FlinkKafkaConsumer011[String](
      "sensor",
      new SimpleStringSchema(),
      properties))
    //    stream3.print()

    //3-youself
    val stream4 = env.addSource(new MySensorSource)
    stream4.print()

    //执行
    env.execute("source test")
  }
}

class MySensorSource() extends SourceFunction[SensorReading] {
  var running: Boolean = true

  override def cancel(): Unit = running = false

  override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = {
    val rand = new Random()
    var curTemp = 1.to(10).map(i => ("sensor_" + i, rand.nextDouble() * 100))

    while (running) {
      curTemp = curTemp.map(
        data => (data._1, data._2 + rand.nextGaussian())
      )
      val curTime = System.currentTimeMillis()
      curTemp.foreach(
        data => sourceContext.collect(SensorReading(data._1, curTime, data._2))
      )
      Thread.sleep(100)
    }
  }

}
